package com.lagou.rpc.client.zookeeper;

import com.alibaba.fastjson.JSONObject;
import com.lagou.rpc.client.Connection;
import com.lagou.rpc.client.ConnectionSource;
import com.lagou.rpc.client.netty.NettyConnectionConfig;
import com.lagou.rpc.client.netty.NettyConnectionSource;
import com.lagou.rpc.common.SimpleRpcConstants;
import com.lagou.rpc.common.entity.HostPortMetaData;
import com.lagou.rpc.common.exceptions.SimpleRpcException;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.util.List;

/**
 * ZookeeperConnectionSource
 *
 * @author wlz
 * @date 2020/5/12
 */
public class ZookeeperConnectionSource implements ConnectionSource {

    private NettyConnectionSource nettyConnectionSource;

    private CuratorFramework client;

    public ZookeeperConnectionSource(ZookeeperConnectionConfig zookeeperConnectionConfig) {
        // 创建netty数据源
        initNettyConnectionSource(zookeeperConnectionConfig);
        // 启动zookeeper客户端
        initZkClient(zookeeperConnectionConfig);
        // 获取服务列表
        loadHostList();
        // 监听服务列表变更事件
        addHostListWatcher();
    }

    private void initNettyConnectionSource(ZookeeperConnectionConfig zookeeperConnectionConfig) {
        NettyConnectionConfig nettyConnectionConfig = new NettyConnectionConfig();
        nettyConnectionConfig.setTimeout(zookeeperConnectionConfig.getTimeout());
        nettyConnectionConfig.setAvailableServerList(null);
        nettyConnectionConfig.setConverter(zookeeperConnectionConfig.getConverter());
        nettyConnectionConfig.setLoadBalanceStrategy(zookeeperConnectionConfig.getLoadBalanceStrategy());
        nettyConnectionConfig.setWorkerThreadSize(zookeeperConnectionConfig.getWorkerThreadSize());
        this.nettyConnectionSource = new NettyConnectionSource(nettyConnectionConfig);
    }

    private void initZkClient(ZookeeperConnectionConfig zookeeperConnectionConfig) {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        client = CuratorFrameworkFactory.builder()
                .connectString(zookeeperConnectionConfig.getZookeeperHost())
                .retryPolicy(retryPolicy)
                // 包含隔离名称
                .namespace(zookeeperConnectionConfig.getNamespace())
                .build();
        client.start();
    }

    private void addHostListWatcher() {
        PathChildrenCache childrenCache = new PathChildrenCache(client, SimpleRpcConstants.SERVER_HOST_LIST_NODE, true);
        PathChildrenCacheListener childrenCacheListener = new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                ChildData data = event.getData();
                HostPortMetaData hostPortMetaData = JSONObject.parseObject(data.getData(), HostPortMetaData.class);
                switch (event.getType()) {
                    case CHILD_ADDED:
                    case CHILD_UPDATED:
                        nettyConnectionSource.putHost(hostPortMetaData);
                        break;
                    case CHILD_REMOVED:
                        nettyConnectionSource.removeHost(hostPortMetaData);
                        break;
                    default:
                        break;
                }
            }
        };
        childrenCache.getListenable().addListener(childrenCacheListener);
        try {
            childrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private void loadHostList() {
        try {
            List<String> serverList = client.getChildren()
                    .forPath(SimpleRpcConstants.SERVER_HOST_LIST_NODE);
            serverList.stream().map(serverNode -> {
                try {
                    byte[] hostContent = client.getData().forPath(SimpleRpcConstants.SERVER_HOST_LIST_NODE + "/" + serverNode);
                    ZookeeperHostPortMetaData hostPortMetaData = JSONObject.parseObject(hostContent, ZookeeperHostPortMetaData.class);
                    hostPortMetaData.setNodeName(serverNode);
                    return hostPortMetaData;
                } catch (Exception e) {
                    throw new SimpleRpcException("get server node data failed", e);
                }
            }).forEach(nettyConnectionSource::putHost);
        } catch (Throwable e) {
            throw new SimpleRpcException("load host list failed ", e);
        }
    }

    @Override
    public Connection getConnection() {
        return nettyConnectionSource.getConnection();
    }
}
